package com.niit.handler

import com.google.gson.Gson
import com.niit.bean.{AdClickData, Answer}
import com.niit.utils.{MyKafkaUtil, SparkUtils}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.streaming.dstream.{DStream, InputDStream}

// 用于接收Kafak中的数据，并封装到相应的样例类中
object DataHandler {

  private val ssc = SparkUtils.takeSSC()

  def kafkaDataHandler(groupId:String,topic:String):DStream[AdClickData]={
    val kafkaData: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(groupId, topic, ssc) //返回的是一批Kafka中的数据
    /*
    1714960532145 华东 深圳 6 2
    1714960532145 华北 上海 6 2
    1714960532145 华南 上海 4 3
     */
    //对这一批数据 进行 封装到样例类中
    val adData: DStream[AdClickData] = kafkaData.map(line => {
      val value: String = line.value()
      val values: Array[String] = value.split(" ")
      AdClickData(values(0), values(1), values(2), values(3), values(4))
    })
    adData
  }

  def kafkaAnswerDataHandler(groupId:String,topic:String):DStream[Answer]={
    //参考代码：将Kafka中的JSON数据封装进样例类
    //  val gson = new Gson   gson.fromJson(kafka数据，classOf[Answer])
    val answerData: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(groupId, topic, ssc)
    val answerRes: DStream[Answer] = answerData.map(kfData => {
      val value: String = kfData.value()
      val gson = new Gson()
      // json 格式的字符串 转换 样例类 要注意的是： json中的字段名 一定要和 样例类中的属性名保持一致
      val answer: Answer = gson.fromJson(value, classOf[Answer])
      answer
    })
    answerRes




  }

  //让ssc开启采集和等待采集
  def startAndAwait(): Unit ={
    ssc.start()
    ssc.awaitTermination()
  }
}
